package com.lmc.log.function;


import com.lmc.log.bean.TrafficFlow;
import net.lmc.util.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class FlowReportWindowFunction implements WindowFunction<TrafficFlow, String, String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow window, Iterable<TrafficFlow> input, Collector<String> out) throws Exception {
        // todo 1 - 获取窗口开始和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // todo 2 - 对窗口中数据进行计算
        TrafficFlow  flowBean = input.iterator().next();
        Long licensePlateCount = flowBean.getLicensePlateCount();

        // todo 3 - 输出结果
        String  output = windowStartTime + "," +
                windowEndTime +"," +
                s + "," +
                licensePlateCount + "," +
                System.currentTimeMillis();

        out.collect(output);
    }
}
